package com.atguigu;

import com.atguigu.bean.UserBean;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;

/**
 * @author yhm
 * @create 2022-12-26 10:16
 */
public class Test05_CSV {
    public static void main(String[] args) {
        // 1. 创建sparkConf配置对象
        SparkConf conf = new SparkConf().setAppName("sql").setMaster("local[*]");

        // 2. 创建sparkSession连接对象
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        // 3. 编写代码
        Dataset<Row> dataset = spark.read()
                .option("header", "true")
                .option("sep", "\t")
                .csv("input/user.csv");

        dataset.show();

        // csv读取的文件  默认所有的类型都是string
        dataset.printSchema();
        // 转换javaBean
//        Dataset<UserBean> beanDataset = dataset.as(Encoders.bean(UserBean.class));
//        beanDataset.show();

        Dataset<UserBean> beanDataset = dataset.map(new MapFunction<Row, UserBean>() {
            @Override
            public UserBean call(Row value) throws Exception {
                return new UserBean(value.getString(0), Long.valueOf(value.getString(1)));
            }
        }, Encoders.bean(UserBean.class));

        beanDataset.show();

        // 写出数据作为csv文件
        beanDataset.write()
                .option("header","true")
                .option("sep",",")
                // 默认输出格式是存在报错
                // append为追加
                // ignore 忽略 如果输出文件夹已经存在 当前写出操作不执行
                // Overwrite 覆盖 删除已经存在的文件夹中的内容 重新写入
                .mode(SaveMode.Overwrite)
                .csv("output");


        // 4. 关闭sparkSession
        spark.close();
    }
}
